package com.leo.ordermessage.comsumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * 特别注意:
 * 1. 线程不是必须要设置为1, 只要使用MessageListenerOrderly顺序监听器即可, 因为同一个Queue的消费由顺序监听器保证
 * 2. 如果使用并行消费监听器 MessageListenerConcurrently, 则必须将线程数设置为1
 */
public class OrderConsumer {
public static void main(String[] args) throws MQClientException {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_11_01");
    consumer.setNamesrvAddr("localhost:9876");

    /**
     * 1. 由于tag使用 * , 所以下面的属性都必须设置为1
     * 2. 另一种方式是使用tag来过滤同一个Topic的不同消息, 即订单每个生命周期的消息都会有一个listener
     *    这样就隔离了不同生命周期的消息并发消费的可能性
     */

    consumer.subscribe("tp_demo_11", "*");

    consumer.setConsumeThreadMin(1);
    consumer.setConsumeThreadMax(1);
    consumer.setPullBatchSize(1);
    consumer.setConsumeMessageBatchMaxSize(1);

    // 从指定的时间戳开始消费
    consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
    // 指定时间戳的值
    consumer.setConsumeTimestamp("2021-01-29 12:00:00");

    /**
     * 使用有序消息监听器, 保证同一个messageQueue中的消息有序消费
     */
    consumer.setMessageListener(new MessageListenerOrderly() {
        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {

            for (MessageExt msg : msgs) {
                System.out.println(
                        msg.getTopic() + "\t" +
                                msg.getQueueId() + "\t" +
                                new String(msg.getBody())
                );
            }

            return ConsumeOrderlyStatus.SUCCESS;
            //return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    });

    consumer.start();

}
}